
import java.io.IOException;
import java.util.Iterator;

import org.apache.commons.math.stat.descriptive.DescriptiveStatistics;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

@SuppressWarnings("deprecation")
public class StatComputation {
	private static class StatsMapper extends MapReduceBase implements
			Mapper<LongWritable, Text, Text, Text> {
		
		private DescriptiveStatistics stats;
		private OutputCollector<Text, Text> outputCollector = null;

		StatsMapper() {
			stats = new DescriptiveStatistics();
		}
		
		@Override
		public void configure(JobConf job) {
		}

		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			if (outputCollector == null) {
				outputCollector = output;
			}
			String[] parts = value.toString().split("\t");
			long count = Long.parseLong(parts[1]);
			stats.addValue(count);
		}
		
		@Override
		public void close() {
			try {
				outputCollector.collect(
						new Text("sum"),
						new Text(stats.getSum() + ":" + stats.getN()));
				outputCollector.collect(
						new Text("sumofsquares"),
						new Text(stats.getSumsq() + ":" + stats.getN()));
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	public static class StatsReducer extends MapReduceBase implements
			Reducer<Text, Text, Text, DoubleWritable> {

		private double mean = 0.0;
		private double sdev = 0.0;
		
		@Override
		public void configure(JobConf job) {
		}

		@Override
		public void reduce(Text key, Iterator<Text> values,
				OutputCollector<Text, DoubleWritable> output, Reporter reporter)
				throws IOException {
			double sum = 0;
			long count = 0;
			while (values.hasNext()) {
				String valueStr = values.next().toString();
				String[] parts = valueStr.split(":");
				sum += Double.parseDouble(parts[0]);
				count += Long.parseLong(parts[1]);
			}
			if (key.toString().compareTo("sum") == 0) {
				mean = sum / count;
				output.collect(new Text("mean"), new DoubleWritable(mean));
			}
			if (key.toString().compareTo("sumofsquares") == 0) {
				double meanOfSquares = sum / count;
				sdev = Math.sqrt(meanOfSquares - (mean * mean));
				output.collect(new Text("sdev"), new DoubleWritable(sdev));
			}
		}
	}

	public static void main(String[] args) throws IOException {

		String inputPath = args[0];
		String outputPath = args[1];

		JobConf conf = new JobConf(StatComputation.class);

		conf.setJobName("StatComputation");
		conf.setNumReduceTasks(1);

		conf.setMapperClass(StatsMapper.class);
		conf.setReducerClass(StatsReducer.class);
		conf.setMapOutputKeyClass(Text.class);
		conf.setMapOutputValueClass(Text.class);
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(DoubleWritable.class);

		FileInputFormat.setInputPaths(conf, new Path(inputPath));
		FileOutputFormat.setOutputPath(conf, new Path(outputPath));

		long startTime = System.currentTimeMillis();
		JobClient.runJob(conf);
		System.out.println("Job Finished in "
				+ (System.currentTimeMillis() - startTime) / 1000.0
				+ " seconds");
	}
}